package com.example.demo.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.retry.RetryPolicy;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

@Configuration
public class KafkaAutoCommitConfig {

    /**
     * 配置自动提交版本的Kafka监听器容器工厂，实现固定1秒间隔的无限重试
     */
    @Bean("autoCommitKafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> autoCommitKafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        
        ConcurrentKafkaListenerContainerFactory<String, String> factory = 
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        
        // 设置确认模式为自动确认
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
        
        // 配置重试模板 - 固定1秒间隔的无限重试
        RetryTemplate retryTemplate = new RetryTemplate();
        
        // 设置重试策略：无限重试
        RetryPolicy retryPolicy = new SimpleRetryPolicy(Integer.MAX_VALUE);
        retryTemplate.setRetryPolicy(retryPolicy);
        
        // 设置固定退避策略：每次重试间隔1秒
        FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
        backOffPolicy.setBackOffPeriod(1000L); // 1秒 = 1000毫秒
        retryTemplate.setBackOffPolicy(backOffPolicy);
        
        factory.setRetryTemplate(retryTemplate);
        
        // 当重试耗尽时的恢复回调（理论上不会被调用，因为是无限重试）
        factory.setRecoveryCallback(context -> {
            System.err.println("自动提交版本 - 重试耗尽（理论上不应该到达这里）: " + context.getLastThrowable().getMessage());
            return null;
        });
        
        return factory;
    }
}
